package examples.bsp;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import util.DiffJobA;
import util.Int;
import util.Utilties;
import bsp.diff.BspDiffHelper;
import bsp.scan.Scan;
import examples.computations.MPS;

public class TestMPS {

	public static class MyMPS extends BspDiffHelper<Int, Int, Int> {
		private static final long serialVersionUID = 1L;

		@Override
		public void setDiffComputation() {
			this.diff = new MPS();
		}

	}

	public static final Log LOG = LogFactory.getLog(TestMPS.class);
	

	public static void main(String[] args) throws InterruptedException,
			IOException, ClassNotFoundException {

		// job configuration

		System.out.println("args.length " + args.length);
		if (args.length >= 2) {
			Path INPUT = new Path(args[0]);
			Path OUTPUT = new Path(args[1]);
			DiffJobA diff = Scan.createScanJob(INPUT, OUTPUT);

			// Set the job name
			diff.setJobName("Diff");
			diff.setJarByClass(TestMPS.class);
			//scan.setBspClass(MyLineofSightDiff.class);
			diff.setBspClass(MyMPS.class);

			
			int outputPrintflg = -1;
			if (args.length > 2)
				diff.setNumBspTask(Integer.parseInt(args[2]));
			else
				diff.setNumBspTask(128);
			if (args.length > 3 && args[3].equalsIgnoreCase("debug")) {
				outputPrintflg = 1;
			}

			System.out.println("total tasks(bsp.setNumBspTask) = "
					+ diff.getNumBspTask());
			long startTime = System.currentTimeMillis();
			if (diff.waitForCompletion(true)) {
				
				System.out.println("Job Finished in "
						+ (System.currentTimeMillis() - startTime) / 1000.0
						+ " seconds");
				// for test
				if (outputPrintflg > 0)
					Utilties.printOutput(diff.getConfiguration(), OUTPUT);
				
			}
			
			FileSystem fs = FileSystem.get(diff.getConfiguration());
			fs.deleteOnExit(OUTPUT);
			
		} else {
			System.err.println("className: INPUT OUTPUT BspTaskNum");
			System.exit(-1);
		}

		System.exit(0);
	}


}